package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.FaultToleranceConstant;
import cn.gwm.flink.streaming.dwm.charge.function.ChargeFunction;
import cn.gwm.flink.streaming.dwm.charge.function.ChargePreFunction;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeConstant;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeEvConstant;
import cn.gwm.flink.streaming.dwm.charge.model.ChargePhevConstant;
import cn.gwm.flink.streaming.dwm.charge.template.ChargeStateModel;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author: liangguang
 * @Date: 2022/10/17
 * @Description: 充电主题测试类
 */
public class DwmChargeTestTask {


    public static void main(String[] args) throws Exception {
        testCharge();
    }

    public static void debugCharge() throws Exception {

        ChargeConstant instance = ChargeEvConstant.getInstance();
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2） 设置相关参数
        env.setParallelism(3);


        String checkpointPath = ConfigLoader.get("hdfsUri") + "/flink/checkpoints/dwmCharge_ev_test";
        BaseTask.envSet(env, checkpointPath);

        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerEarLiest(
                "consumer-dwm-charge-ev-test1",
                ConfigLoader.get("kafka.clean.topic.dwd.ev", "dwd_clean_ev")
        );

        OutputTag<JSONObject> summarStream = new OutputTag<JSONObject>("summar") {
        };
        OutputTag<JSONObject> detailStream = new OutputTag<JSONObject>("detail") {
        };
        SingleOutputStreamOperator<JSONObject> process = env.addSource(kafkaConsumer)
                .map(JSONUtil::parseObj)
                .disableChaining()
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .flatMap(new ChargePreFunction(instance))
                .disableChaining()
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .flatMap(new ChargeFunction(instance))
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject chargeData, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (ChargeConstant.detail.equals(chargeData.getStr("type"))) {
                            ctx.output(detailStream, chargeData);
                        } else {
                            ctx.output(summarStream, chargeData);
                        }
                    }
                });

        DataStream<JSONObject> summarProucerStream = process.getSideOutput(summarStream)
                .map(item -> {
                    JSONObject transJson = BeanSource.transJson(item, BeanSource.SourceEnum.evChargeSummar);
                    if (item.containsKey(ChargeStateModel.END_FLAG)) {
                        transJson.set(ChargeStateModel.END_FLAG, BaseFields.defaultVal);
                    }
                    transJson.set("favourable", item.get("favourable") == null ? false : item.get("favourable"));
                    return transJson;
                });

        summarProucerStream.filter(item -> item.containsKey(ChargeStateModel.END_FLAG))
                .addSink(BaseTask.getFileSink(ConfigLoader.get("hive.table.name.dwm.charge.summar"),
                        ".txt",
                        "dwm/charge_test",
                        BeanSource.SourceEnum.evChargeSummar))
                .name("toHdfsSummar-ev-test1");

        env.execute("dwmCharge-ev");


    }


    public static void testCharge() throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2） 设置相关参数
        env.setParallelism(1);
        BaseTask.envSet(env, ConfigLoader.get(FaultToleranceConstant.FLINK_JOB_NAME));
        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(
                "consumer-test",
                "dwm_charge_test"
        );
        OutputTag<JSONObject> summarStream = new OutputTag<JSONObject>("summar") {
        };
        OutputTag<JSONObject> detailStream = new OutputTag<JSONObject>("detail") {
        };
        OutputTag<JSONObject> favourableStream = new OutputTag<JSONObject>("favourable") {
        };
        DataStream<JSONObject> sourceDataStream = env.addSource(kafkaConsumer).map(JSONUtil::parseObj);

        SingleOutputStreamOperator<JSONObject> evFilter = sourceDataStream.filter(item -> item.containsKey("custom_type") && "ev".equals(item.getStr("custom_type").toLowerCase()));
        SingleOutputStreamOperator<JSONObject> phevFilter = sourceDataStream.filter(item -> item.containsKey("custom_type") && "phev".equals(item.getStr("custom_type").toLowerCase()));


        SingleOutputStreamOperator<JSONObject> evProcess = evFilter
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .flatMap(new ChargePreFunction(ChargeEvConstant.getInstance()))
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .flatMap(new ChargeFunction(ChargeEvConstant.getInstance()))
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject chargeData, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (ChargeConstant.detail.equals(chargeData.getStr("type"))) {
                            ctx.output(detailStream, chargeData);
                        } else if (ChargeConstant.favourable.equals(chargeData.getStr("type"))) {
                            ctx.output(favourableStream, chargeData);
                        } else {
                            ctx.output(summarStream, chargeData);
                        }
                    }
                });

        DataStream<JSONObject> evSummarProucerStream = evProcess.getSideOutput(summarStream)
                .map(item -> {
                    JSONObject transJson = BeanSource.transJson(item, BeanSource.SourceEnum.evChargeSummar);
                    if (item.containsKey(ChargeStateModel.END_FLAG)) {
                        transJson.set(ChargeStateModel.END_FLAG, BaseFields.defaultVal);
                    }
                    transJson.set("favourable", item.get("favourable") == null ? false : item.get("favourable"));
                    return transJson;
                });
        DataStream<JSONObject> favourableProcucerStream = evProcess.getSideOutput(favourableStream);

        evSummarProucerStream.addSink(FlinkKafkaUtil.toKafka("dwm_charge_summar")).name("toKafkaSummar-test");

//        evSummarProucerStream.filter(item -> item.containsKey(ChargeStateModel.END_FLAG))
//                .addSink(BaseTask.getFileSink(ConfigLoader.get("hive.table.name.dwm.charge.summar"),
//                        ".txt",
//                        "dwm/" + ConfigLoader.get("hive.table.name.dwm.charge.summar"),
//                        BeanSource.SourceEnum.evChargeSummar))
//                .name("toHdfsSummarTest");

        favourableProcucerStream.addSink(BaseTask.getFileSink("favourable",
                        ".txt",
                        "dwm/ev_favourable",
                        BeanSource.SourceEnum.chargeFavourable))
                .name("toHdfsEvFavourable");


        SingleOutputStreamOperator<JSONObject> phevProcess = phevFilter
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .flatMap(new ChargePreFunction(ChargePhevConstant.getInstance()))
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .flatMap(new ChargeFunction(ChargePhevConstant.getInstance()))
                .process(new ProcessFunction<JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject chargeData, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (ChargeConstant.detail.equals(chargeData.getStr("type"))) {
                            ctx.output(detailStream, chargeData);
                        } else if (ChargeConstant.favourable.equals(chargeData.getStr("type"))) {
                            ctx.output(favourableStream, chargeData);
                        } else {
                            ctx.output(summarStream, chargeData);
                        }
                    }
                });

        DataStream<JSONObject> phevSummarProucerStream = phevProcess.getSideOutput(summarStream)
                .map(item -> {
                    JSONObject transJson = BeanSource.transJson(item, BeanSource.SourceEnum.evChargeSummar);
                    if (item.containsKey(ChargeStateModel.END_FLAG)) {
                        transJson.set(ChargeStateModel.END_FLAG, BaseFields.defaultVal);
                    }
                    transJson.set("favourable", item.get("favourable") == null ? false : item.get("favourable"));
                    return transJson;
                });

        phevSummarProucerStream.addSink(FlinkKafkaUtil.toKafka("dwm_charge_summar")).name("toKafkaSummar-test2");

        DataStream<JSONObject> favourableProcucerStream2 = phevProcess.getSideOutput(favourableStream);

        favourableProcucerStream2.addSink(BaseTask.getFileSink("favourable",
                        ".txt",
                        "dwm/phev_favourable",
                        BeanSource.SourceEnum.chargeFavourable))
                .name("toHdfsPhevFavourable");

//        phevSummarProucerStream.filter(item -> item.containsKey(ChargeStateModel.END_FLAG))
//                .addSink(BaseTask.getFileSink(ConfigLoader.get("hive.table.name.dwm.charge.summar"),
//                        ".txt",
//                        "dwm/" + ConfigLoader.get("hive.table.name.dwm.charge.summar"),
//                        BeanSource.SourceEnum.evChargeSummar))
//                .name("toHdfsSummarTest2");

        env.execute("charge-test");

    }

}
